package com.heima.kafka.sample;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * kafka信息生产者
 */
public class ProducerQuickStart {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //1:kafka配置
        Properties prop = new Properties();
        //1.1:kafka的链接地址
        //ProducerConfig.BOOTSTRAP_SERVERS_CONFIG = "bootstrap.servers"
        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");

        //1.2：key和value的序列化
        prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");

        //配置ack : 消息发送机制： 默认：1 ：只要集群首领leads收到信息，生产者就会收到发送成功响应
        prop.put(ProducerConfig.ACKS_CONFIG,1);
        //配置：retries  消息发送重试机制 每次重试 10ms
        prop.put(ProducerConfig.RETRIES_CONFIG,10);
        //配置：消息压缩，降低网络传输和储存开销 :snappy , lz4 ,gzip 压缩方式
        prop.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");


        //2:创建kafka消息生产对象
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(prop);

        //3:发送消息
        /**
         * 1:topic
         * 2:消息的 key
         * 3:消息的 value
         */
        ProducerRecord<String, String> producerRecord =
                new ProducerRecord<String, String>("itcast-topic-input", "key-demo", "Hello ItZheng");

        //发送信息两种方式： 同步异步： 通常使用异步
        //区别： 同步一个参数：消息参数， 异步：两个参数： 消息参数和回调接口对象new Callback()
        // 2：大量消息发送：  异步会更快一些

        long begin1 = System.currentTimeMillis();
        for (int i = 0; i < 1000; i++) {
            //同步发送 //答应偏移量： +  .get() .offset()
            long offset = producer.send(producerRecord).get().offset();
        }
        long end1 = System.currentTimeMillis();
        System.out.println("同步耗时："+(end1 - begin1));


        long begin2 = System.currentTimeMillis();
        for (int i = 0; i < 1000; i++) {
            //异步发送：
            producer.send(producerRecord, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e!=null){
                        System.out.println("出现异常："+e);
                    }
                    //打印偏移量
                    System.out.println(recordMetadata.offset());
                }
            });
        }
        long end2 = System.currentTimeMillis();
        System.out.println("异步耗时："+(end2 - begin2));


        //4:关闭消息通道,必须关闭，否则消息发送不成功
        producer.close();
    }
}
